package com.mjf.app.function;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;

/**
 * 自定义 FlinkCDC 反序列化方式
 */
public class CustomerDeserialization implements DebeziumDeserializationSchema<String> {
    /**
     * 将数据封装为 Json 格式返回
     *
     * @param sourceRecord 来源数据
     * @param collector    发送器
     * @throws Exception {
     *                   "database":"",
     *                   "tableName":"",
     *                   "before":{"":"", "":""...},
     *                   "after":{"":"", "":""...},
     *                   "type":"c u d"
     *                   }
     */
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
        // 1.创建 Json 对象用于存储最终数据
        JSONObject result = new JSONObject();

        // 2.获取库名和表名
        String topic = sourceRecord.topic();
        String database = topic.split("\\.")[1];
        String tableName = topic.split("\\.")[2];

        Struct value = (Struct) sourceRecord.value();

        // 3.获取 before 数据
        Struct before = value.getStruct("before");
        JSONObject beforeJson = new JSONObject();
        if (before != null) {    // 新增数据没有 before
            Schema beforeSchema = before.schema();
            List<Field> beforeFields = beforeSchema.fields();   // 获取 before 中的所有字段
            for (Field beforeField : beforeFields) {
                Object beforeValue = before.get(beforeField);
                beforeJson.put(beforeField.name(), beforeValue);
            }
        }

        // 4.获取 after 数据
        Struct after = value.getStruct("after");
        JSONObject afterJson = new JSONObject();
        if (after != null) {    // 删除数据没有 after
            Schema afterSchema = after.schema();
            List<Field> afterFields = afterSchema.fields();   // 获取 after 中的所有字段
            for (Field afterField : afterFields) {
                Object afterValue = after.get(afterField);
                afterJson.put(afterField.name(), afterValue);
            }
        }

        // 5.获取操作类型 CREATE UPDATE DELETE
        Envelope.Operation operation = Envelope.operationFor(sourceRecord);
        String type = operation.toString().toLowerCase();
        if ("create".equals(type)) {
            type = "insert";    // 对 CREATE 进行映射，方便后续使用
        }

        // 6.将字段写入 Json 对象
        result.put("database", database);
        result.put("tableName", tableName);
        result.put("before", beforeJson);
        result.put("after", afterJson);
        result.put("type", type);

        // 7.输出数据
        collector.collect(result.toJSONString());
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
}
